package consul

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/rs/zerolog/log"
	"io"
	"math/rand"
	"net/http"
	"sync"
	"time"
)

type PolicyConfig struct {
	WaitTime      time.Duration
	Timeout       time.Duration
	RetryMaxDelay time.Duration
	Client        *http.Client
}

func NewPolicy(retryMaxDelay, timeout, waitTime int) *PolicyConfig {
	return &PolicyConfig{
		WaitTime:      time.Duration(waitTime) * time.Second,
		Timeout:       time.Duration(timeout) * time.Second,
		RetryMaxDelay: time.Duration(retryMaxDelay) * time.Second,
		Client:        &http.Client{Timeout: 90 * time.Second},
	}
}

func defaultConfig() *PolicyConfig {
	return &PolicyConfig{
		WaitTime:      60 * time.Second,
		Timeout:       90 * time.Second,
		RetryMaxDelay: 10 * time.Second,
		Client:        &http.Client{Timeout: 90 * time.Second},
	}
}

// ServiceDiscovery 基于 Consul 的服务发现客户端
type ServiceDiscovery struct {
	client      *http.Client
	consulAddr  string
	serviceName string

	mu        sync.RWMutex
	instances []map[string]interface{} // 缓存的健康实例列表

	nextMu    sync.Mutex
	nextIndex int // 轮询索引

	lastIndex string // 上次 Consul Index，用于长轮询

	ctx    context.Context
	cancel context.CancelFunc
}

// NewServiceDiscovery 创建服务发现客户端并启动长轮询
func NewServiceDiscovery(consulAddr, serviceName string, cfg *PolicyConfig) *ServiceDiscovery {
	if cfg == nil {
		cfg = defaultConfig()
	}

	client := cfg.Client
	if client == nil {
		client = &http.Client{Timeout: cfg.Timeout}
	}

	ctx, cancel := context.WithCancel(context.Background())

	sd := &ServiceDiscovery{
		client:      client,
		consulAddr:  consulAddr,
		serviceName: serviceName,
		lastIndex:   "0",
		ctx:         ctx,
		cancel:      cancel,
	}

	go sd.blockingPoll(cfg.WaitTime)
	return sd
}

// Stop 停止服务发现监听
func (sd *ServiceDiscovery) Stop() {
	sd.cancel()
}

// blockingPoll 使用 Consul 长轮询监听服务实例变化
func (sd *ServiceDiscovery) blockingPoll(waitTime time.Duration) {
	const maxBodySize = 1 << 20 // 1MB
	baseRetryDelay := 500 * time.Millisecond
	retryDelay := baseRetryDelay

	for {
		select {
		case <-sd.ctx.Done():
			return
		default:
		}

		// 重试前等待（首次不等待）
		if retryDelay > baseRetryDelay {
			sleepWithJitter(sd.ctx, retryDelay)
			retryDelay = retryDelay * 2
			if retryDelay > 10*time.Second {
				retryDelay = 10 * time.Second
			}
		}

		// 构造请求 URL
		url := fmt.Sprintf("%s/v1/health/service/%s?passing&wait=%ds&index=%s",
			sd.consulAddr, sd.serviceName, int(waitTime.Seconds()), sd.getLastIndex())

		req, err := http.NewRequestWithContext(sd.ctx, "GET", url, nil)
		if err != nil {
			log.Printf("event=blocking_poll_error service=%s error=failed_to_create_request err=%v", sd.serviceName, err)
			continue
		}

		resp, err := sd.client.Do(req)
		if err != nil {
			log.Printf("event=blocking_poll_error service=%s error=request_failed err=%v", sd.serviceName, err)
			continue
		}

		// 延迟关闭 body
		bodyReader := io.LimitReader(resp.Body, maxBodySize+1)
		body, err := io.ReadAll(bodyReader)
		defer resp.Body.Close()

		// 检查状态码
		if resp.StatusCode != http.StatusOK {
			truncatedBody := string(body)
			if len(truncatedBody) > 512 {
				truncatedBody = truncatedBody[:512]
			}
			log.Printf("event=blocking_poll_error service=%s error=bad_status status=%s body=%s",
				sd.serviceName, resp.Status, truncatedBody)
			continue
		}

		// 读取 X-Consul-Index
		newIndex := resp.Header.Get("X-Consul-Index")
		if newIndex == "" {
			log.Printf("event=blocking_poll_error service=%s error=missing_consul_index", sd.serviceName)
			continue
		}

		// 检查响应体是否超限
		if len(body) > maxBodySize {
			log.Printf("event=blocking_poll_error service=%s error=response_too_large size=%d", sd.serviceName, len(body))
			continue
		}

		// 判断是否有更新
		if newIndex == sd.getLastIndex() {
			// 无变更（wait 超时），仅更新 index
			sd.setLastIndex(newIndex)
			log.Printf("event=no_change service=%s index=%s", sd.serviceName, newIndex)
			retryDelay = baseRetryDelay // 成功响应，重置重试
			continue
		}

		// 有变更：解析 body
		var instances []map[string]interface{}
		if err := json.Unmarshal(body, &instances); err != nil {
			log.Printf("event=blocking_poll_error service=%s error=decode_failed err=%v body=%.512s",
				sd.serviceName, err, string(body))
			continue
		}

		// 更新状态：先更新数据，再更新 index
		sd.setInstancesAndIndex(instances, newIndex)
		log.Printf("event=instances_updated service=%s index=%s count=%d",
			sd.serviceName, newIndex, len(instances))

		// 成功处理变更，重置重试延迟
		retryDelay = baseRetryDelay
	}
}

// GetOneInstance 返回一个健康实例（轮询策略）
func (sd *ServiceDiscovery) GetOneInstance() (map[string]interface{}, error) {
	sd.mu.RLock()
	instances := make([]map[string]interface{}, len(sd.instances))
	copy(instances, sd.instances)
	sd.mu.RUnlock()

	if len(instances) == 0 {
		return nil, fmt.Errorf("ServerDiscovery.GetOneInstance: no healthy instances available for service %s", sd.serviceName)
	}

	sd.nextMu.Lock()
	idx := sd.nextIndex % len(instances)
	sd.nextIndex++
	sd.nextMu.Unlock()

	return instances[idx], nil
}

// GetAllInstances 返回所有健康实例的副本（只读）
func (sd *ServiceDiscovery) GetAllInstances() []map[string]interface{} {
	sd.mu.RLock()
	defer sd.mu.RUnlock()

	instances := make([]map[string]interface{}, len(sd.instances))
	copy(instances, sd.instances)
	return instances
}

// sleepWithJitter 带随机抖动的 sleep，避免大量客户端同时重试
func sleepWithJitter(ctx context.Context, delay time.Duration) {
	jitter := time.Duration(rand.Int63n(int64(delay / 2))) // 0 ~ delay/2 的随机抖动
	select {
	case <-time.After(delay + jitter):
	case <-ctx.Done():
	}
}

// 封装并发安全的 lastIndex 操作
func (sd *ServiceDiscovery) getLastIndex() string {
	sd.mu.Lock()
	defer sd.mu.Unlock()
	return sd.lastIndex
}

func (sd *ServiceDiscovery) setLastIndex(index string) {
	sd.mu.Lock()
	defer sd.mu.Unlock()
	sd.lastIndex = index
}

func (sd *ServiceDiscovery) setInstancesAndIndex(instances []map[string]interface{}, index string) {
	sd.mu.Lock()
	defer sd.mu.Unlock()
	sd.instances = instances
	sd.lastIndex = index
}
